package com.gmadmin.mq.template;

import cn.hutool.json.JSONUtil;
import com.gmadmin.common.BaseMqMessage;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.spring.annotation.ExtRocketMQTemplateConfiguration;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.apache.rocketmq.spring.support.RocketMQHeaders;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;

/**
 * @author zhangliang
 * @version 1.0
 * @date 2022/7/27 23:39
 */
//读取自定义配置
//@ExtRocketMQTemplateConfiguration(nameServer = "${demo.rocketmq.extNameServer}", tlsEnable = "${demo.rocketmq.ext.useTLS}", instanceName = "pztest33")
@ExtRocketMQTemplateConfiguration
public class CustomMessageTemplate extends RocketMQTemplate {

    private static final Logger LOGGER = LoggerFactory.getLogger(CustomMessageTemplate.class);


    /**
     * springboot不支持使用header传递tags，根据要求，需要在topic后进行拼接 formats: `topicName:tags`，不拼接标识无tag
     */
    public String buildDestination(String topic, String tag) {
        return StringUtils.isBlank(tag) ? topic : topic + ":" + tag;
    }


    /**
     * 发送同步消息
     *
     * @param topic   主题
     * @param tag     标签
     * @param message 消息
     * @param <T>     泛型
     * @return SendResult
     */
    public <T extends BaseMqMessage> SendResult syncSend(String topic, String tag, T message) {
        // 设置业务键，此处根据公共的参数进行处理
        // 更多的其它基础业务处理...
        String destination = buildDestination(topic, tag);
        Message<T> sendMessage = MessageBuilder.withPayload(message).setHeader(RocketMQHeaders.KEYS, message.getKey()).build();
        SendResult sendResult = syncSend(destination, sendMessage);
        // 此处为了方便查看给日志转了json，根据选择选择日志记录方式，例如ELK采集
        LOGGER.info("[{}]同步消息[{}]发送结果[{}]", destination, JSONUtil.toJsonStr(message), JSONUtil.toJsonStr(sendResult));
        return sendResult;
    }

    /**
     * 发送同步延迟消息
     *
     * @param topic      主题
     * @param tag        标签
     * @param message    消息
     * @param delayLevel -1是否立即执行
     * @param <T>        泛型
     * @return SendResult
     */
    public <T extends BaseMqMessage> SendResult syncSendDelay(String topic, String tag, T message, int delayLevel) {
        String destination = buildDestination(topic, tag);
        Message<T> sendMessage = MessageBuilder.withPayload(message).setHeader(RocketMQHeaders.KEYS, message.getKey()).build();
        SendResult sendResult = syncSend(destination, sendMessage, 3000, delayLevel);
        LOGGER.info("[{}]延迟等级[{}]消息[{}]发送结果[{}]", destination, delayLevel, JSONUtil.toJsonStr(message), JSONUtil.toJsonStr(sendResult));
        return sendResult;
    }


    /**
     * 发送事务消息
     *
     * @param topic       主题
     * @param tag         标签
     * @param message     消息体
     * @param headerValue key
     * @param <T>         泛型
     * @return result
     */
    public <T extends BaseMqMessage> SendResult sendMessageForTransaction(String topic, String tag, T message, String headerValue) {
        String destination = buildDestination(topic, tag);
        Message msg = MessageBuilder.withPayload(message).setHeader(RocketMQHeaders.TRANSACTION_ID, headerValue).build();
        SendResult sendResult = sendMessageInTransaction(destination, msg, null);
        LOGGER.info("-----[{}]---rocketMQTemplate send Transactional msg body = [{}] , sendResult=[{}]", destination, msg.getPayload(), sendResult.getSendStatus());
        return sendResult;
    }
}
